package com.shujia.sql

import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

object Demo1WordCount {
  def main(args: Array[String]): Unit = {
    /**
     * 在新版本的spark中，如果想要编写spark sql的话，需要使用新的spark入口类：SparkSession
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("wc spark sql")
      .getOrCreate()

    /**
     * spark sql和spark core的核心数据类型不太一样
     *
     * 1、读取数据构建一个DataFrame,相当于一张表
     */
    val linesDF: DataFrame = sparkSession.read
      .format("csv") //指定读取数据的格式
      .schema("line STRING") //指定列的名和列的类型，多个列之间使用,分割
      .option("sep", "\n") //指定分割符，csv格式读取默认是英文逗号
      .load("spark/data/words.txt") // 指定要读取数据的位置，可以使用相对路径

//    println(linesDF)
//    linesDF.show() //查看DF中的数据内容（表内容）
//    linesDF.printSchema() //查看DF表结构

    /**
     * 2、DF本身是无法直接在上面写sql的，需要将DF注册成一个视图，才可以写sql数据分析
     */
    linesDF.createOrReplaceTempView("lines") // 起一个表名，后面的sql语句可以做查询分析

    /**
     * 3、可以编写sql语句 （统计单词的数量）
     * spark sql是完全兼容hive sql
     */
    val resDF: DataFrame = sparkSession.sql(
      """
        |select
        |t1.word as word,
        |count(1) as counts
        |from
        |(select
        | explode(split(line,'\\|')) as word from lines) t1
        | group by t1.word
        |""".stripMargin)

    /**
     * 4、将计算的结果DF保存到HDFS上
     */
    val resDS: Dataset[Row] = resDF.repartition(1)
    resDS.write
      .format("csv") //指定输出数据文件格式
      .option("sep","\t") // 指定列之间的分隔符
      .mode(SaveMode.Overwrite) // 使用SaveMode枚举类，设置为覆盖写
      .save("spark/data/sqlout1") // 指定输出的文件夹

  }
}
